package com.example.artemis;

import com.example.artemis.config.EmbeddedServerConfig;
import com.example.artemis.config.ApplicationConfig;
import com.example.artemis.service.MessageConsumerService;
import com.example.artemis.service.MessageProducerService;
import com.example.artemis.statistics.StatisticsService;
import com.example.artemis.statistics.DefaultStatisticsService;
import com.example.artemis.statistics.StatisticsEventListener;
import com.example.artemis.events.EventBus;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.jms.ConnectionFactory;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Main application for ActiveMQ Artemis embedded message queuing
 *
 * Features:
 * - Embedded ActiveMQ Artemis server with persistent storage
 * - Queue sharding across multiple queues using hashCode() % queueCount
 * - Multi-threaded producers (configurable parallel producers)
 * - Multi-threaded consumers (configurable parallel consumers)
 * - At-least-once delivery semantics with manual acknowledgment
 * - Message persistence and redelivery
 * - Event-driven statistics collection
 * - Decoupled statistics service
 * - Centralized configuration management
 */
public class ActiveMQApplication {
    private static final Logger logger = LoggerFactory.getLogger(ActiveMQApplication.class);

    private final EmbeddedServerConfig serverConfig;
    private final ConnectionFactory connectionFactory;
    private final AtomicReference<MessageProducerService> producerService = new AtomicReference<>();
    private final AtomicReference<MessageConsumerService> consumerService = new AtomicReference<>();
    private final StatisticsService statisticsService;

    public ActiveMQApplication() {
        // 验证配置参数
        ApplicationConfig.validateConfig();

        this.serverConfig = new EmbeddedServerConfig();
        this.connectionFactory = createConnectionFactory();
        this.statisticsService = new DefaultStatisticsService();

        // 初始化统计服务
        initializeStatistics();

        // 注册JVM关闭钩子
        registerShutdownHook();

        logger.info("应用程序初始化完成");
    }

    /**
     * 初始化统计服务
     */
    private void initializeStatistics() {
        try {
            // 启动统计服务
            statisticsService.start();

            // 注册统计事件监听器到事件总线
            StatisticsEventListener statisticsListener = new StatisticsEventListener(statisticsService);
            EventBus.getInstance().registerGlobalListener(statisticsListener);

            logger.info("统计服务初始化完成，事件监听器已注册");
        } catch (Exception e) {
            logger.error("统计服务初始化失败", e);
            throw new RuntimeException("统计服务初始化失败", e);
        }
    }

    /**
     * 注册JVM关闭钩子
     */
    private void registerShutdownHook() {
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            logger.info("JVM关闭钩子被触发，执行紧急清理...");
            try {
                // 紧急关闭所有资源
                if (statisticsService != null && statisticsService.isRunning()) {
                    statisticsService.stop();
                }

                EventBus.getInstance().shutdownNow();

                if (serverConfig.isRunning()) {
                    serverConfig.stop();
                }

                logger.info("紧急资源清理完成");
            } catch (Exception e) {
                logger.error("紧急资源清理时发生错误", e);
            }
        }, "ShutdownHook"));
    }

    /**
     * Create JMS ConnectionFactory directly
     */
    private ConnectionFactory createConnectionFactory() {
        try {
            // Create connection factory for embedded server
            return new ActiveMQConnectionFactory("vm://0");
        } catch (Exception e) {
            throw new RuntimeException("Failed to create ConnectionFactory", e);
        }
    }

    /**
     * Main entry point for the application
     */
    public static void main(String[] args) {
        ActiveMQApplication application = new ActiveMQApplication();

        try {
            application.run();
        } catch (Exception e) {
            logger.error("Application failed", e);
            System.exit(1);
        }
    }

    /**
     * Run the complete application
     */
    public void run() throws Exception {
        logger.info("=".repeat(80));
        logger.info("STARTING {} v{}", ApplicationConfig.APP_NAME, ApplicationConfig.APP_VERSION);
        logger.info("=".repeat(80));
        logger.info("配置信息:");
        logger.info("{}", ApplicationConfig.getAllConfigInfo());
        logger.info("=".repeat(80));

        try {
            // Start embedded server
            startEmbeddedServer();

            // Start consumers first (ready to receive messages)
            startConsumers();

            // Give consumers time to initialize
            Thread.sleep(ApplicationConfig.CONSUMER_STARTUP_WAIT_TIME);

            // Start producers
            startProducers();

            // Monitor and display statistics
            monitorDemo();

            // Shutdown gracefully
            shutdown();

        } catch (Exception e) {
            logger.error("Application failed", e);
            throw e;
        }

        logger.info("=".repeat(80));
        logger.info("APPLICATION COMPLETED");
        logger.info("=".repeat(80));
    }

    /**
     * Start the embedded ActiveMQ Artemis server
     */
    private void startEmbeddedServer() throws Exception {
        logger.info("Starting embedded ActiveMQ Artemis server...");

        serverConfig.createServer();
        serverConfig.start();

        logger.info("Embedded server started successfully");
        logger.info("Queue configuration:");
        for (String queueName : EmbeddedServerConfig.QUEUE_NAMES) {
            logger.info("  - {}", queueName);
        }
        logger.info("Persistence settings:");
        logger.info("  - Max disk usage: {}%", EmbeddedServerConfig.MAX_DISK_USAGE_PERCENTAGE);
        logger.info("  - Max queue size: {}MB", EmbeddedServerConfig.MAX_QUEUE_SIZE_BYTES / (1024 * 1024));
        logger.info("  - Journal sync timeout: {}ms", EmbeddedServerConfig.JOURNAL_SYNC_TIMEOUT);
    }



    /**
     * Start the consumer service
     */
    private void startConsumers() {
        logger.info("Starting consumer service...");

        // 使用共享的统计服务实例
        MessageConsumerService consumers = new MessageConsumerService(statisticsService);

        consumerService.set(consumers);

        // Start consumers asynchronously
        consumers.startConsumers()
                .thenAccept(result -> logger.info("Consumer service completed: {}", result))
                .exceptionally(throwable -> {
                    logger.error("Consumer service failed", throwable);
                    return null;
                });

        // Wait for consumers to be ready
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        logger.info("Consumer service started with {} consumers", ApplicationConfig.NUMBER_OF_CONSUMERS);
    }

    /**
     * Start the producer service
     */
    private void startProducers() {
        logger.info("Starting producer service...");

        // 使用共享的统计服务实例
        MessageProducerService producers = new MessageProducerService(statisticsService);

        producerService.set(producers);

        // Start producers and wait for completion
        producers.startProducers()
                .thenAccept(result -> {
                    logger.info("Producer service completed: {}", result);
                    logger.info("All messages have been sent to queues");
                })
                .exceptionally(throwable -> {
                    logger.error("Producer service failed", throwable);
                    return null;
                });

        logger.info("Producer service started with {} producers", ApplicationConfig.NUMBER_OF_PRODUCERS);
    }

    /**
     * Monitor the demo and display statistics
     */
    private void monitorDemo() throws InterruptedException {
        logger.info("Monitoring demo for {} seconds...", ApplicationConfig.DEMO_DURATION_SECONDS);

        long startTime = System.currentTimeMillis();
        long lastStatsTime = startTime;

        while (System.currentTimeMillis() - startTime < ApplicationConfig.DEMO_DURATION_SECONDS * 1000) {
            Thread.sleep(ApplicationConfig.STATISTICS_DISPLAY_INTERVAL_SECONDS * 1000);

            long currentTime = System.currentTimeMillis();
            long elapsedSeconds = (currentTime - lastStatsTime) / 1000;

            // Display current statistics
            displayStatistics(elapsedSeconds);
            lastStatsTime = currentTime;
        }
    }

    /**
     * Display current statistics with structured output
     */
    private void displayStatistics(long elapsedSeconds) {
        MessageProducerService producers = producerService.get();
        MessageConsumerService consumers = consumerService.get();

        // 如果没有任何服务在运行，不显示统计信息
        if (producers == null && consumers == null) {
            return;
        }

        logger.info("=".repeat(90));
        logger.info("实时统计信息 - 监控时间窗口: {} 秒", elapsedSeconds);
        logger.info("统计时间: {}", java.time.LocalDateTime.now().format(java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        logger.info("=".repeat(90));

        // 显示生产者统计
        if (producers != null) {
            var producerStats = producers.getStats();
            logger.info("📤 生产者性能指标:");
            logger.info("   ├─ 发送总量: {} 条消息", formatNumber(producerStats.totalSent()));
            logger.info("   ├─ 失败数量: {} 条消息", formatNumber(producerStats.totalFailed()));
            logger.info("   ├─ 重试次数: {} 次", formatNumber(producerStats.totalRetried()));
            logger.info("   ├─ 平均发送延迟: {} ms", formatDecimal(producerStats.averageProductionTime()));
            logger.info("   └─ 生产完成度: {}", producers.getProductionProgress());

            // 计算成功率
            long totalProducerAttempts = producerStats.totalSent() + producerStats.totalFailed();
            if (totalProducerAttempts > 0) {
                double producerSuccessRate = (double) producerStats.totalSent() / totalProducerAttempts * 100;
                logger.info("   └─ 发送成功率: {}%", String.format("%.2f", producerSuccessRate));
            }
        } else {
            logger.info("📤 生产者状态: ⏸️ 未启动");
        }

        logger.info(""); // 空行分隔

        // 显示消费者统计
        if (consumers != null) {
            var consumerStats = consumers.getStats();
            logger.info("📥 消费者性能指标:");
            logger.info("   ├─ 处理总量: {} 条消息", formatNumber(consumerStats.totalProcessed()));
            logger.info("   ├─ 失败数量: {} 条消息", formatNumber(consumerStats.totalFailed()));
            logger.info("   ├─ 重试次数: {} 次", formatNumber(consumerStats.totalRetried()));
            logger.info("   ├─ 平均处理延迟: {} ms", formatDecimal(consumerStats.averageProcessingTime()));

            // 计算成功率和吞吐量
            long totalConsumerAttempts = consumerStats.totalProcessed() + consumerStats.totalFailed();
            if (totalConsumerAttempts > 0) {
                double consumerSuccessRate = (double) consumerStats.totalProcessed() / totalConsumerAttempts * 100;
                double throughput = (double) totalConsumerAttempts / elapsedSeconds; // 每秒处理消息数
                logger.info("   ├─ 处理成功率: {}%", String.format("%.2f", consumerSuccessRate));
                logger.info("   └─ 平均吞吐量: {} 消息/秒", String.format("%.2f", throughput));
            }
        } else {
            logger.info("📥 消费者状态: ⏸️ 未启动");
        }

        // 显示系统整体状态
        displaySystemHealth();

        logger.info("=".repeat(90));
    }

    /**
     * 显示系统健康状态
     */
    private void displaySystemHealth() {
        logger.info(""); // 空行分隔
        logger.info("🖥️ 系统健康状态:");

        // 内存使用情况
        Runtime runtime = Runtime.getRuntime();
        long totalMemory = runtime.totalMemory();
        long freeMemory = runtime.freeMemory();
        long usedMemory = totalMemory - freeMemory;
        long maxMemory = runtime.maxMemory();

        double memoryUsagePercent = (double) usedMemory / maxMemory * 100;
        logger.info("   ├─ 内存使用: {} / {} ({})%",
                   formatBytes(usedMemory), formatBytes(maxMemory), String.format("%.1f", memoryUsagePercent));

        // JVM信息
        logger.info("   ├─ JVM运行时间: {} 秒",
                   java.lang.management.ManagementFactory.getRuntimeMXBean().getUptime() / 1000);

        // ActiveMQ服务器状态
        if (serverConfig.isRunning()) {
            logger.info("   └─ ActiveMQ服务器: 🟢 运行中");
        } else {
            logger.info("   └─ ActiveMQ服务器: 🔴 已停止");
        }
    }

    /**
     * 格式化数字显示
     */
    private String formatNumber(long number) {
        if (number >= 1_000_000) {
            return String.format("%.1fM", number / 1_000_000.0);
        } else if (number >= 1_000) {
            return String.format("%.1fK", number / 1_000.0);
        } else {
            return String.format("%d", number);
        }
    }

    /**
     * 格式化小数显示
     */
    private String formatDecimal(double number) {
        if (number == 0) {
            return "0";
        } else if (number < 1) {
            return String.format("%.3f", number);
        } else if (number < 100) {
            return String.format("%.2f", number);
        } else {
            return String.format("%.1f", number);
        }
    }

    /**
     * 格式化字节数显示
     */
    private String formatBytes(long bytes) {
        if (bytes >= 1_073_741_824) {
            return String.format("%.1f GB", bytes / 1_073_741_824.0);
        } else if (bytes >= 1_048_576) {
            return String.format("%.1f MB", bytes / 1_048_576.0);
        } else if (bytes >= 1_024) {
            return String.format("%.1f KB", bytes / 1_024.0);
        } else {
            return String.format("%d B", bytes);
        }
    }

    /**
     * Shutdown the application gracefully
     */
    private void shutdown() {
        logger.info("开始优雅关闭应用程序...");

        try {
            // 1. 停止生产者服务（停止发送新消息）
            MessageProducerService producers = producerService.get();
            if (producers != null) {
                logger.info("正在停止生产者服务...");
                producers.stop();
            }

            // 2. 等待现有消息被处理完成
            logger.info("等待剩余消息处理完成...");
            Thread.sleep(ApplicationConfig.MESSAGE_PROCESSING_WAIT_TIME);

            // 3. 停止消费者服务
            MessageConsumerService consumers = consumerService.get();
            if (consumers != null) {
                logger.info("正在停止消费者服务...");
                consumers.stop();
            }

            // 4. 停止嵌入式服务器
            logger.info("正在停止嵌入式ActiveMQ服务器...");
            serverConfig.stop();

            // 5. 显示最终统计信息
            displayFinalStatistics();

            // 6. 停止统计服务
            if (statisticsService != null) {
                logger.info("正在停止统计服务...");
                statisticsService.stop();
            }

            // 7. 关闭事件总线
            logger.info("正在关闭事件总线...");
            try {
                EventBus.getInstance().shutdown();
                logger.info("事件总线已关闭");
            } catch (Exception e) {
                logger.warn("关闭事件总线时发生错误", e);
            }

            logger.info("应用程序优雅关闭完成");

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.warn("关闭过程被中断");
        } catch (Exception e) {
            logger.error("关闭过程中发生错误", e);
        }
    }

    /**
     * Display final statistics summary
     */
    private void displayFinalStatistics() {
        logger.info("=".repeat(100));
        logger.info("最终统计报告");
        logger.info("=".repeat(100));

        MessageProducerService producers = producerService.get();
        MessageConsumerService consumers = consumerService.get();

        // 分别统计生产者和消费者信息
        displayProducerStatistics(producers);
        displayConsumerStatistics(consumers);

        // 如果两者都存在，显示综合分析
        if (producers != null && consumers != null) {
            displayCombinedAnalysis(producers, consumers);
        }

        // 显示队列状态
        displayQueueStatus();

        // 显示统计服务详细信息
        displayStatisticsServiceDetails();

        logger.info("=".repeat(100));
    }

    /**
     * 显示生产者统计信息
     */
    private void displayProducerStatistics(MessageProducerService producers) {
        if (producers != null) {
            logger.info("生产者总结:");
            var producerStats = producers.getStats();
            logger.info("  - 总发送消息数: {}", producerStats.totalSent());
            logger.info("  - 总失败数: {}", producerStats.totalFailed());
            logger.info("  - 总重试数: {}", producerStats.totalRetried());
            logger.info("  - 平均发送时间: {}ms", String.format("%.2f", producerStats.averageProductionTime()));
            logger.info("  - 预期消息数: {}", ApplicationConfig.NUMBER_OF_PRODUCERS * ApplicationConfig.MESSAGES_PER_PRODUCER);
            logger.info("  - 生产进度: {}", producers.getProductionProgress());
        } else {
            logger.info("生产者状态: 未启动");
        }
    }

    /**
     * 显示消费者统计信息
     */
    private void displayConsumerStatistics(MessageConsumerService consumers) {
        if (consumers != null) {
            logger.info("消费者总结:");
            var consumerStats = consumers.getStats();
            logger.info("  - 总处理消息数: {}", consumerStats.totalProcessed());
            logger.info("  - 总失败数: {}", consumerStats.totalFailed());
            logger.info("  - 总重试数: {}", consumerStats.totalRetried());
            logger.info("  - 平均处理时间: {}ms", String.format("%.2f", consumerStats.averageProcessingTime()));

            long total = consumerStats.totalProcessed() + consumerStats.totalFailed();
            logger.info("  - 总处理数: {}", total);

            if (total > 0) {
                long successRate = consumerStats.totalProcessed() * 100 / total;
                logger.info("  - 整体成功率: {}%", successRate);
            }
        } else {
            logger.info("消费者状态: 未启动");
        }
    }

    /**
     * 显示综合分析（当生产者和消费者都存在时）
     */
    private void displayCombinedAnalysis(MessageProducerService producers, MessageConsumerService consumers) {
        logger.info("综合分析:");
        var producerStats = producers.getStats();
        var consumerStats = consumers.getStats();

        if (producerStats.totalSent() > 0) {
            long processingRate = consumerStats.totalProcessed() * 100 / producerStats.totalSent();
            logger.info("  - 相对于生产数的处理率: {}%", processingRate);
        }
    }

    
    /**
     * 显示队列状态
     */
    private void displayQueueStatus() {
        logger.info("队列状态:");
        for (int i = 0; i < EmbeddedServerConfig.NUMBER_OF_QUEUES; i++) {
            logger.info("  - {}: {}", EmbeddedServerConfig.QUEUE_PREFIX + i, "活跃");
        }
    }

    /**
     * 显示统计服务详细信息
     */
    private void displayStatisticsServiceDetails() {
        logger.info("统计服务详细信息:");

        if (statisticsService != null) {
            // 显示统计服务摘要
            logger.info(statisticsService.getStatisticsSummary());

            // 显示全局统计详细信息
            var globalStats = statisticsService.getGlobalStatistics();
            logger.info("全局统计详细:");
            logger.info("  - 消息发送: {}", globalStats.getTotalSent());
            logger.info("  - 消息处理: {}", globalStats.getTotalProcessed());
            logger.info("  - 处理失败: {}", globalStats.getTotalFailed());
            logger.info("  - 重试次数: {}", globalStats.getTotalRetried());
            logger.info("  - 平均生产时间: {} ms", String.format("%.2f", globalStats.getAverageProductionTime()));
            logger.info("  - 平均处理时间: {} ms", String.format("%.2f", globalStats.getAverageProcessingTime()));
            logger.info("  - 成功率: {}%", String.format("%.2f", globalStats.getSuccessRate()));
            logger.info("  - 完成率: {}%", String.format("%.2f", globalStats.getCompletionRate()));

            // 显示队列统计
            var queueStats = statisticsService.getAllQueueStatistics();
            if (!queueStats.isEmpty()) {
                logger.info("队列统计详情:");
                queueStats.forEach((queueName, stats) -> {
                    logger.info("  {}: 发送={}, 处理={}, 失败={}, 成功率={}%",
                            queueName, stats.getSentCount(), stats.getProcessedCount(),
                            stats.getFailedCount(), String.format("%.1f", stats.getSuccessRate()));
                });
            }
        } else {
            logger.info("统计服务: 未初始化");
        }
    }
}